package my.rpc.service;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import my.rpc.common.bean.RpcRequest;
import my.rpc.common.bean.RpcResponse;
import my.rpc.common.codec.RpcDecoder;
import my.rpc.common.codec.RpcEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 扫描并开启服务
 * @author fengbo
 * @date 2017/11/2
 */
public class RpcServer implements ApplicationContextAware, InitializingBean {

    private static final Logger log = LoggerFactory.getLogger(RpcServer.class);

    private String serviceAddress;

    /**
     * key 存放的是自定义注解 RpcService 里的 value 的 name
     * value 存储的是被注解的类
     */
    private Map<String, Object> handlerMap = new ConcurrentHashMap<>(128);

    /***
     * 通过 ApplicationContext 获取被 RpcService 注解的类，并添加到 handlerMap 中，
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

        // 从applicationContext中取出所有被RpcService的类名和类
        Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(RpcService.class);
        if (serviceBeanMap == null || serviceBeanMap.isEmpty()) {
            return;
        }

        for (Object serviceBean : serviceBeanMap.values()) {
            RpcService rpcService = serviceBean.getClass().getAnnotation(RpcService.class);
            String serviceName = rpcService.value().getName();
            handlerMap.put(serviceName, serviceBean);
        }
    }

    /***
     * spring 初始化完成后执行
     */
    @Override
    public void afterPropertiesSet() {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) {
                            ChannelPipeline pipeline = channel.pipeline();
                            // 粘包处理
                            pipeline.addLast(new LengthFieldBasedFrameDecoder(65536,
                                    // 长度字段的偏差，从第几个字节开始表示数据长度
                                    0,
                                    // 长度字段占的字节数
                                    4,
                                    // 添加到长度字段的补偿值
                                    0,
                                    // 从解码帧中第一次去除的字节数
                                    0));
                            pipeline.addLast(new RpcDecoder<>(RpcRequest.class));
                            pipeline.addLast(new RpcEncoder<>(RpcResponse.class));
                            pipeline.addLast(new RpcServerHandler(handlerMap));
                        }
                    });
            // 获取服务端的地址和端口号
            String[] addressArray = StringUtils.split(serviceAddress, ":");
            String host = addressArray[0];
            int port = Integer.parseInt(addressArray[1]);
            log.info("service started on port {}", port);
            bootstrap.bind(host, port).syncUninterruptibly()
                    .channel().closeFuture().syncUninterruptibly();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    public void setServiceAddress(String serviceAddress) {
        this.serviceAddress = serviceAddress;
    }
}
